#!/usr/bin/env python
# -*- coding: UTF-8 -*-
'''
@Project ：bignet_sumo_dev
@File ：main-dev-tongluoshan.py
@IDE ：PyCharm
@Author ：xinyingjie
@Date ：2025/3/24 9:40
'''

from __future__ import absolute_import
from __future__ import print_function

import asyncio
import math
import queue
import json
import threading
import random
import time
from multiprocessing import Process, Queue
from kafka import KafkaConsumer
from config.config import KAFKA_HOST
from loguru import logger
from websocket.websocketserver1 import WebSocketServer

logger.add("debug.log", format="{time} {level} {message}", level="DEBUG")

KAFKA_E1_LIST = ['CJ-10-SUM', 'JJ-10-SUM']


def senf_kafka_E1(lidar_queue_data):
    ws_server = WebSocketServer()
    ws_server.start()
    loop = asyncio.new_event_loop()  # 创建新的事件循环
    asyncio.set_event_loop(loop)
    send_time = 0
    send_count = 0
    frame_num = 0
    while True:
        try:
            t1 = time.time()
            frame_num += 1
            simu_json_list_before = {'obj': []}
            # print(lidar_queue_data.qsize())
            while not lidar_queue_data.empty():
                lidar_json_list = lidar_queue_data.get()
                if len(lidar_json_list) > 0:
                    simu_json_list_before['obj'] += list(lidar_json_list)

                future = asyncio.run_coroutine_threadsafe(
                    ws_server.broadcast(str(simu_json_list_before).replace("'", '"')),
                    ws_server.loop)
                future.result()  # 等待协程完成
            end_time = time.time()
            if end_time - send_time > 1:
                logger.info(f"发送帧率 {abs(send_count - frame_num)}")
                send_time = end_time
                send_count = frame_num = 0
            if end_time - t1 < 0.1:
                time.sleep(0.1 - (end_time - t1))
        except Exception as e:
            logger.error(f"发送异常{str(e)}")
            time.sleep(0.1)
            continue


def read_trafficE1_channel(target_channel, lidar_queue_data):
    # 实时接收全域E1帧数据
    kafka_consumer_E1 = KafkaConsumer(
        target_channel,
        bootstrap_servers=KAFKA_HOST,
        auto_offset_reset='latest',
        enable_auto_commit=False
    )
    logger.info(f"全域E1 {target_channel} 监听已就绪")
    while True:
        time.sleep(0.1)
        for message in kafka_consumer_E1:
            if message:
                try:
                    one_list = []
                    message_value = message.value.decode('utf-8')
                    data = json.loads(message_value)
                    if 'targetList' not in data.keys():
                        continue
                    # print(data)
                    e1FrameParticipant_list = data['targetList']
                    ts = int(time.time() * 1000)
                    # ts = get_timestamp_milliseconds_jingtai(data['globalTime'])
                    for one in e1FrameParticipant_list:
                        sourceId = one['station']
                        one['sourceId'] = sourceId
                        temp = {"id": one['id'], "type": 1, 'source': 4,
                                "laneNum": 0,
                                "longitude": one['lon'], "latitude": one['lat'], "speed": int(30 * 3.6),
                                "courseAngle": round(one['angle'], 1), 'plate': "默A00000", "flag": 1,
                                'orgCode': one['orgCode']
                                }
                        one_list.append(temp)
                    lidar_queue_data.put(one_list)
                except Exception as e:
                    logger.error(str(e))
                    continue


def main_loop():
    lidar_queue_data = Queue(maxsize=2000)  # 多进程队列在此处创建
    for KAFKA_E1 in KAFKA_E1_LIST:
        son_process = Process(target=read_trafficE1_channel,
                              args=(KAFKA_E1, lidar_queue_data), daemon=True)
        son_process.start()
    senf_kafka_E1(lidar_queue_data)


if __name__ == '__main__':
    main_loop()
